package org.example.config;

import org.aopalliance.intercept.MethodInterceptor;
import org.example.annotation.ConsumerKafkaMessage;
import org.example.annotation.ProducerKafkaMessage;
import org.example.consumer.LocalKafkaConsumer;
import org.example.listener.B;
import org.springframework.aop.Advisor;
import org.springframework.aop.AfterAdvice;
import org.springframework.aop.AfterReturningAdvice;
import org.springframework.aop.MethodBeforeAdvice;
import org.springframework.aop.framework.adapter.AdvisorAdapter;
import org.springframework.aop.framework.adapter.AdvisorAdapterRegistry;
import org.springframework.aop.framework.adapter.UnknownAdviceTypeException;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor;
import org.springframework.beans.factory.support.GenericBeanDefinition;
import org.springframework.context.event.EventListener;
import org.springframework.core.MethodIntrospector;
import org.springframework.core.annotation.AnnotatedElementUtils;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.util.CollectionUtils;
import org.youshang.service.impl.HelloServiceImpl;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.GregorianCalendar;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author youshang
 * @date 2021/02/20 10:59
 * 用于拦截当前项目中有多少方法在使用KafkaMessage注解
 **/
public class KafkaMessageBeanPostProcessor implements BeanPostProcessor, BeanDefinitionRegistryPostProcessor {

    ConcurrentHashMap<String, List<String>> topicMap = new ConcurrentHashMap<>();
    ConcurrentHashMap<Object, Method> producerMethodMap = new ConcurrentHashMap<>();

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        Method[] methods = bean.getClass().getMethods();
        Object targetObject = bean;
        for (Method method : methods) {
            boolean consumerKafkaMessageAnnotation = method.isAnnotationPresent(ConsumerKafkaMessage.class);
            if (consumerKafkaMessageAnnotation) {
                //发送消息调用的方法
                consumerMethod(method, targetObject);
            }

            boolean producerKafkaMessageAnnotation = method.isAnnotationPresent(ProducerKafkaMessage.class);
            if (producerKafkaMessageAnnotation) {
                //发送消息调用的方法
                producerMethod(method, targetObject);
//                producerMethodMap.put(targetObject,method);
            }

        }

        return bean;
    }

    /**
     * 接受消息调用的方法
     *
     * @param method
     */
    private void consumerMethod(Method method, Object targetObject) {
        ConsumerKafkaMessage annotation = AnnotationUtils.getAnnotation(method, ConsumerKafkaMessage.class);
        String topic = annotation.topic();
        String consumerGroup = annotation.consumerGroup();
        List<String> consumerGroupList = topicMap.get(topic);
        if (null == consumerGroupList || consumerGroupList.size() == 0) {
            List<String> list = new ArrayList<>();
            list.add(consumerGroup);
            topicMap.put(topic, list);
        } else {
            consumerGroupList.add(consumerGroup);
            topicMap.put(topic, consumerGroupList);
        }
        if (null != topicMap && topicMap.size() != 0) {
            consumerGroupList = topicMap.get(topic);
            //根据每个消费组创建消费者线程
            for (String group : consumerGroupList) {
                LocalKafkaConsumer localKafkaConsumer = new LocalKafkaConsumer(topic, group, targetObject, method);
                //TODO 需要改线程调用方式
                Thread thread = new Thread(localKafkaConsumer);
                thread.start();
            }
        }
    }

    /**
     * 发送消息调用的方法
     *
     * @param method
     */
    private void producerMethod(Method method, Object targetObject) {
        ProducerKafkaMessage annotation = AnnotationUtils.getAnnotation(method, ProducerKafkaMessage.class);
        String topic = annotation.topic();

        try {
//            afterReturning(method.getReturnType(), method, method.getParameters(), targetObject);
        } catch (Throwable throwable) {
            throwable.printStackTrace();
        }
    }


    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {

    }

    @Override
    public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {
        GenericBeanDefinition genericBeanDefinition = new GenericBeanDefinition();
        genericBeanDefinition.setBeanClass(B.class);
        registry.registerBeanDefinition("b",genericBeanDefinition);
    }
}
